package com.celeritymq.session;

import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.jms.JMSException;
import javax.jms.TemporaryTopic;
import javax.jms.Topic;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
import com.celeritymq.consumer.CelerityTopicSubscriber;
import com.celeritymq.destination.CelerityTemporaryTopic;
import com.celeritymq.destination.CelerityTopic;
import com.celeritymq.producer.CelerityTopicPublisher;
import com.celeritymq.utils.CelerityDestinationNameGenerator;

public class CelerityTopicSession extends CeleritySession implements TopicSession {
	private static final String NAME = "TOPIC";
	private CelerityTopicSubscriber durableSubscriber;
	private Map<String, List<CelerityTopicPublisher>> publisherMap = new HashMap<String, List<CelerityTopicPublisher>>();
	private Map<String, List<CelerityTopicSubscriber>> subscriberMap = new HashMap<String, List<CelerityTopicSubscriber>>();
	private Map<String, CelerityTemporaryTopic> temporaryTopicMap = new HashMap<String, CelerityTemporaryTopic>();
	private Map<String, CelerityTopic> topicMap = new HashMap<String, CelerityTopic>();

	public CelerityTopicSession(SocketAddress clientAddress, boolean transacted, int acknowledgeMode) {
		super(clientAddress);
		this.transacted = transacted;
		this.acknowledgeMode = acknowledgeMode;
	}

	@Override
	public TopicSubscriber createDurableSubscriber(Topic topic, String topicName) throws JMSException {
		return createDurableSubscriber(topic, topicName, null, false);
	}

	@Override
	public TopicSubscriber createDurableSubscriber(Topic topic, String topicName, String messageSelector, boolean noLocal)
			throws JMSException {
		if (topic.getClass() != CelerityTopic.class) {
			throw new JMSException("not celeritytopic");
		}
		durableSubscriber = new CelerityTopicSubscriber((CelerityTopic) topic, messageSelector, noLocal);
		return durableSubscriber;
	}

	@Override
	public TopicPublisher createPublisher(Topic topic) throws JMSException {
		if (topic.getClass() != CelerityTopic.class) {
			throw new JMSException("not celeritytopic");
		}
		CelerityTopicPublisher ctp = new CelerityTopicPublisher((CelerityTopic) topic);
		List<CelerityTopicPublisher> publisherList = publisherMap.get(topic);
		if (publisherList == null) {
			publisherList = new ArrayList<CelerityTopicPublisher>();
			publisherMap.put(topic.getTopicName(), publisherList);
		}
		publisherList.add(ctp);
		return ctp;
	}

	@Override
	public TopicSubscriber createSubscriber(Topic topic) throws JMSException {
		return createSubscriber(topic, null, false);
	}

	@Override
	public TopicSubscriber createSubscriber(Topic topic, String messageSelector, boolean noLocal) throws JMSException {
		if (topic.getClass() != CelerityTopic.class) {
			throw new JMSException("not celeritytopic");
		}
		CelerityTopicSubscriber subscriber = new CelerityTopicSubscriber((CelerityTopic) topic, messageSelector, noLocal);
		List<CelerityTopicSubscriber> subscriberList = subscriberMap.get(topic.getTopicName());
		if (subscriberList == null) {
			subscriberList = new ArrayList<CelerityTopicSubscriber>();
			subscriberMap.put(topic.getTopicName(), subscriberList);
		}
		subscriberList.add(subscriber);
		return subscriber;
	}

	@Override
	public TemporaryTopic createTemporaryTopic() throws JMSException {
		String tempTopicName = null;
		do {
			tempTopicName = CelerityDestinationNameGenerator.generateName(NAME);
		} while (tempTopicName == null || !temporaryTopicMap.containsKey(tempTopicName));
		CelerityTemporaryTopic ctt = new CelerityTemporaryTopic(this, tempTopicName);
		temporaryTopicMap.put(tempTopicName, ctt);
		return ctt;
	}

	@Override
	public Topic createTopic(String topicName) throws JMSException {
		if (topicMap.containsKey(topicName)) {
			return topicMap.get(topicName);
		}
		CelerityTopic ct = new CelerityTopic(topicName);
		topicMap.put(topicName, ct);
		return ct;
	}

	@Override
	public void unsubscribe(String topicName) throws JMSException {
		topicMap.remove(topicName);
	}

	public void unsubscribeTemporaryTopic(String topicName) {
		temporaryTopicMap.remove(topicName);
	}

}
